package com.z.redis.aof;

import com.z.redis.RedisCore;
import com.z.redis.command.Command;
import com.z.redis.command.WriteCommand;
import com.z.redis.resp.Resp;
import com.z.redis.resp.RespArray;
import com.z.redis.util.PropertiesUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Method;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import com.z.redis.util.Format;
import com.z.redis.command.CommandFactory;

import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.RandomAccess;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class Aof {
    private static final org.apache.log4j.Logger LOGGER = org.apache.log4j.Logger.getLogger(Aof.class);

    private static final String suffix = ".aof";

    /**
     * 1.经过大量测试 使用过3年以上机械磁盘的最大性能为26
     * 2.存盘偏移量 控制单个持久化文件大小
     */
    public static final int shiftBit = 26;// 文件分段偏移
    private Long aofPutIndex = 0L;// 当前写入偏移量
    private String fileName = PropertiesUtil.getAofPath();// AOF文件路径
    private RingBlockingQueue<Resp> runtimeRespQueue = new RingBlockingQueue<Resp>(8888, 888888);// 内存命令队列
    private ByteBuf bufferPolled = PooledByteBufAllocator.DEFAULT.buffer(8888, Integer.MAX_VALUE);// 池化内容缓冲区
    private ScheduledThreadPoolExecutor persistenceExecutor = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r,"Aof_Single_Thread");
        }
    });
    private final RedisCore redisCore;
    // 读写锁
    private final ReadWriteLock reentrantLock = new ReentrantReadWriteLock();

    // 确保AOF文件目录存在 避免写入失败
    public Aof(RedisCore redisCore) {
        this.redisCore = redisCore;
        File file = new File(this.fileName + suffix);
        if(!file.isDirectory()){
            File parentFile = file.getParentFile();
            if(parentFile != null || !parentFile.exists()){
                parentFile.mkdirs(); // 创建目录
            }
        }
        start(); //启动持久化任务
    }
    // 将非阻塞写入队列
    public void put(Resp resp) { runtimeRespQueue.offer(resp); }
    public void start(){
        /**
         * 谁先执行需要按照异步顺序
         */
        persistenceExecutor.execute(this::pickupDiskDataAllSegment);// 启动时加载历史AOF数据
        persistenceExecutor.scheduleAtFixedRate(this::downDiskAllSegment, 10, 1, TimeUnit.SECONDS);// 每1秒持久化一次
    }
    public void close(){
        try{
            persistenceExecutor.shutdown();
        }catch (Exception e){
            LOGGER.warn("Exception",e);
        }
    }
    /**
     * 定时持久化方法
     * 面向过程分段存储所有的数据
     * 将内存中的命令批量写入分段AOF文件，动态扩展文件大小（64MB分段 + 8MB扩展步长）
     */
    public void downDiskAllSegment(){
        reentrantLock.writeLock().lock(); // 阻塞直到获取锁
        try {
            long segmentId = -1;
            Segment:
            while (segmentId!=(aofPutIndex>>shiftBit)) {
                //要后28位
                segmentId=(aofPutIndex>>shiftBit);
                // 读取文件
                RandomAccessFile randomAccessFile = new RandomAccessFile(fileName+"_"+segmentId+suffix, "rw");
                FileChannel channel= randomAccessFile.getChannel();
                long len=channel.size();

                int putIndex= Format.uintNBit(aofPutIndex,shiftBit); // 段内偏移量（最低26位）
                long baseOffset=aofPutIndex-putIndex;// 全局偏移量起点

                if (len-putIndex < 1L <<(shiftBit-2)){ // 剩余空间不足16MB时扩展
                    len = segmentId + 1 <<(shiftBit-2); // 扩展成16MB倍数
                }
                // 内存映射
                MappedByteBuffer mappedByteBuffer =channel.map(FileChannel.MapMode.READ_WRITE, 0,len);

                while(true){
                    Resp resp= runtimeRespQueue.peek(); // 查看队列头部命令
                    if(resp==null){
                        //bufferPolled.release();
                        clean(mappedByteBuffer);
                        randomAccessFile.close();
                        break Segment;
                    }
                    Resp.write(resp,bufferPolled); // 序列化命令到缓冲区
                    int respLen=bufferPolled.readableBytes();
                    if((mappedByteBuffer.capacity()<=respLen+putIndex)){ // 空间不足扩展映射
                        len += 1L <<(shiftBit-3); // 每次拓展8MB
                        // 重新映射
                        mappedByteBuffer = channel.map(FileChannel.MapMode.READ_WRITE, 0,len);
                        if(len>(1<<shiftBit)){
                            bufferPolled.release();
                            aofPutIndex=baseOffset+1<<shiftBit;
                            break;
                        }
                    }
                    while(respLen>0){ // 逐字节写入文件
                        mappedByteBuffer.put(putIndex++,bufferPolled.readByte());
                        respLen--;
                    }
                    /**
                     * 完成消费
                     */
                    aofPutIndex= baseOffset+putIndex; // 更新全局偏移量
                    runtimeRespQueue.poll(); // 移除已处理命令
                    bufferPolled.clear(); // 清空缓冲区
                    if(len-putIndex < (1L <<(shiftBit-3))){ // 剩余空间不足8MB再次扩展
                        len+= 1L <<(shiftBit-3);
                        if(len>(1<<shiftBit)){
                            bufferPolled.release();
                            clean(mappedByteBuffer);
                            aofPutIndex=baseOffset+1<<shiftBit;
                            break ;
                        }
                        mappedByteBuffer = channel.map(FileChannel.MapMode.READ_WRITE, 0,len);
                    }
                }
            }
        } catch (Exception e) {
            System.err.println(e.getMessage());
            LOGGER.error("aof Exception",e);
        }
        finally {
            reentrantLock.writeLock().unlock();
        }
    }

    /**
     * 从AOF文件分段加载历史命令，反序列化后执行以恢复Redis内存状态
     */
    public void pickupDiskDataAllSegment()  {
        reentrantLock.writeLock().lock();
        try {
            long segmentId=-1;
            Segment:
            while (segmentId!=(aofPutIndex>>shiftBit)) { // 遍历所有段文件
                //要后28位
                segmentId=(aofPutIndex>>shiftBit);
                RandomAccessFile randomAccessFile = new RandomAccessFile(fileName+"_"+segmentId+suffix, "r");
                FileChannel channel= randomAccessFile.getChannel();
                long len=channel.size();
                int putIndex= Format.uintNBit(aofPutIndex,shiftBit) ;
                long baseOffset=aofPutIndex-putIndex;

                MappedByteBuffer mappedByteBuffer =channel.map(FileChannel.MapMode.READ_ONLY, 0,len);
                ByteBuf bufferPolled= new PooledByteBufAllocator().buffer((int) len);
                bufferPolled.writeBytes(mappedByteBuffer);// 加载全部文件内容到缓冲区

                while (true){
                    Resp resp=null;
                    try {
                        resp=Resp.decode(bufferPolled); // 反序列化命令
                    }catch (Exception e) { // 解码失败退出
                        clean(mappedByteBuffer);
                        randomAccessFile.close();
                        bufferPolled.release();
                        break Segment;
                    }
                    Command command = CommandFactory.from((RespArray) resp); // 生成命令对象
                    WriteCommand writeCommand=   (WriteCommand)command; // 强制转换为写命令
                    assert writeCommand != null;
                    writeCommand.handle(this.redisCore); // 执行命令恢复数据
                    putIndex=bufferPolled.readerIndex(); // 获取当前读取位置
                    aofPutIndex=putIndex+ baseOffset; // 更新全局偏移量
                    if(putIndex>(1<<shiftBit)){ // 超过段大小（64MB）时跳转
                        bufferPolled.release();
                        clean(mappedByteBuffer);
                        aofPutIndex=baseOffset+1<<shiftBit;
                        break ;
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            reentrantLock.writeLock().unlock();
        }
    }

    public static void clean(final MappedByteBuffer buffer) throws Exception {
        if (buffer == null) {
            return;
        }
        buffer.force();
        //Privileged特权
        AccessController.doPrivileged((PrivilegedAction<Object>) () -> {
            try {
                // System.out.println(buffer.getClass().getName());
                Method getCleanerMethod = buffer.getClass().getMethod("cleaner", new Class[0]);
                getCleanerMethod.setAccessible(true);
                sun.misc.Cleaner cleaner = (sun.misc.Cleaner) getCleanerMethod.invoke(buffer, new Object[0]);
                cleaner.clean();
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        });
    }
}
